package cn.com.infcn.spider.io.input.impl;

import java.net.UnknownHostException;
import java.util.Collections;
import java.util.List;
import java.util.Map;

import org.nlpcn.commons.lang.util.MD5;

import com.alibaba.fastjson.JSONObject;
import com.mongodb.BasicDBObject;
import com.mongodb.DBCursor;
import com.mongodb.DBObject;

import cn.com.infcn.spider.URIManager;
import cn.com.infcn.spider.io.input.DBIteartorInput;
import cn.com.infcn.spider.util.MongoDao;
import cn.com.infcn.spider.util.StaticValue;

public class mongoInput implements DBIteartorInput {

	private MongoDao mongoDao = null;

	private String tableName = null;

	private int batchSize = 0;

	private Object flag = null;
	
	private Object type = null;	//1:分页 2：增量

	private sortMethod comp;

	private String flagName;

	private String id;

	private String jobName;

	public mongoInput(JSONObject job, String jobName)
			throws UnknownHostException {

		tableName = job.getString("tableName");

		batchSize = job.getIntValue("batchSize");

		comp = sortMethod.valueOf(job.getString("sortMethod"));

		flagName = job.getString("flagName");
		
		type = job.getString("type");

		if (tableName == null) {
			throw new RuntimeException("not set tableName! in "
					+ job.toJSONString().replace("\n", " "));
		}

		mongoDao = new MongoDao(job.getString("host"), job.getInteger("port"),
				job.getString("dbName"));

		this.jobName = jobName;

		this.id = MD5.code(job.getString("host") + job.getInteger("port")
				+ job.getString("dbName") + tableName + flagName);

		DBObject sortValue = StaticValue.getBean(MongoDao.class, "mongoDao")
				.findById(URIManager.SUCCESS + jobName, id);

		if (sortValue != null) {
			this.flag = sortValue.get("flag");
		} else {
			this.flag = job.get("flag");
		}

	}

	@Override
	public List<?> all() {
		BasicDBObject query = null;

		if (comp != null && flagName != null) {
			query = new BasicDBObject("$" + comp, new BasicDBObject(flagName,
					flag));
		}
		return mongoDao.findByKV(tableName, query);
	}

	private String lastCode;

	@Override
	public List<?> iterator() {

		
		
		DBCursor limit = null;
		try {
			//分页
			if("1".equals(this.type)){
				limit = mongoDao.getCollection(tableName).find()
					.sort(new BasicDBObject(flagName, 1)).skip((Integer)this.flag).limit(batchSize);
			}
			//增量
			else{
				DBObject query = new BasicDBObject(flagName, new BasicDBObject("$"
						+ comp.name().toLowerCase(), this.flag));
				
				limit = mongoDao.getCollection(tableName).find(query)
						.sort(new BasicDBObject(flagName, 1)).limit(batchSize);
			}
			List<DBObject> array = limit.toArray();

			if (array == null || array.size() <= 0) {
				return Collections.emptyList();
			}

			DBObject last = array.get(array.size() - 1);

			String tempCode = last.toString();

			if (tempCode.equals(lastCode)) {
				return Collections.emptyList();
			}
			lastCode = tempCode;

			changeFlag((BasicDBObject) last, array.size());

			return array;
		}finally{
			limit.close();
		}
	}

	public void changeFlag( Map<String, Object> obj, int step) {
		Object _flag = null;
		Object _newFlag = null;
		//分页
		if("1".equals(this.type)){
			this.flag = (Integer)this.flag + step;
		}
		//增量
		else{
			if (comp == null || flagName == null) {
				return;
			}
			
			Object newFlag = obj.get(flagName);
			
			if (newFlag == null) {
				return;
			}
			
			if (newFlag.equals(flag)) {
				return;
			}
			
			if(flag.getClass() != newFlag.getClass()){
				_flag = flag.toString();
				_newFlag = newFlag.toString();
			}else{
				_flag = flag;
				_newFlag = newFlag;
			}
			
			if (sortMethod.GT.equals(comp) || sortMethod.GTE.equals(comp)) {
				if (((Comparable) _flag).compareTo(_newFlag) < 0) {
					this.flag = newFlag;
				}
			} else {
				if (((Comparable) _flag).compareTo(_newFlag) > 0) {
					this.flag = newFlag;
				}
			}
		}

		BasicDBObject dbo = new BasicDBObject("_id", id).append("flag", this.flag);

		StaticValue.getBean(MongoDao.class, "mongoDao").save(
				URIManager.SUCCESS + jobName, dbo);
	}

	@Override
	public sortMethod getComp() {
		return comp;
	}

	@Override
	public String getFlagName() {
		return flagName;
	}

	@Override
	public String getId() {
		return id;
	}

	public Object getFlag() {
		return flag;
	}

	public void setFlag(Object flag) {
		this.flag = flag;
	}

}
